package tuplestore

import (
	"fmt"
	"math"
	"time"

	"github.com/cespare/xxhash"
)

func (ts *TupleStore) Insert(tuple *Tuple) error {
	if tuple == nil {
		return ErrNilTuple
	}
	if tuple.ID == "" {
		return ErrEmptyTupleID
	}

	start := time.Now()
	defer func() {
		ts.metrics.InsertDuration.Observe(time.Since(start).Seconds())
		ts.metrics.InsertCount.Inc()
	}()

	ts.mu.Lock()
	defer ts.mu.Unlock()

	if _, _, exists := ts.idIndex.LookupID(tuple.ID); exists {
		return fmt.Errorf("%w: %q", ErrDuplicateID, tuple.ID)
	}

	segment, block, err := ts.selectOptimalLocation(tuple)
	if err != nil {
		return fmt.Errorf("failed to select location: %w", err)
	}

	block.DeltaStore.AddTuple(tuple)
	ts.updateIndices(tuple, segment.ID, block.ID)

	if ts.shouldMerge(block) {
		go ts.scheduleBlockMerge(segment.ID, block.ID)
	}

	return nil
}

func (ts *TupleStore) BatchInsert(tuples []*Tuple) error {
	if len(tuples) == 0 {
		return nil
	}

	// 从第一个元组获取Schema
	schema := tuples[0].Schema

	ts.mu.Lock()
	defer ts.mu.Unlock()

	// 简单负载均衡：找到元组数量最少的段和块
	var targetSegment *Segment
	var targetBlock *Block
	minTuples := math.MaxInt32

	for _, segment := range ts.segments {
		for _, block := range segment.Blocks {
			block.mu.RLock()
			deltaCount := len(block.DeltaStore.tuples)
			mainCount := len(block.MainStore.tuples)
			totalCount := deltaCount + mainCount
			block.mu.RUnlock()

			if totalCount < minTuples {
				minTuples = totalCount
				targetSegment = segment
				targetBlock = block
			}
		}
	}

	if targetBlock == nil {
		return fmt.Errorf("无法找到可用的区块")
	}

	// 设置块的Schema
	targetBlock.mu.Lock()
	if len(targetBlock.Schema) == 0 {
		targetBlock.Schema = schema
	}
	targetBlock.mu.Unlock()

	// 批量添加元组
	for _, tuple := range tuples {
		// 检查ID是否已存在
		_, _, exists := ts.idIndex.LookupID(tuple.ID)
		if exists {
			continue // 跳过已存在的元组
		}

		// 添加到目标块
		targetBlock.DeltaStore.AddTuple(tuple)

		// 更新索引
		ts.idIndex.Insert(tuple.ID, targetSegment.ID, targetBlock.ID)

		// 更新指标
		ts.metrics.InsertCount.Inc()
	}

	return nil
}

func (ts *TupleStore) selectOptimalLocation(tuple *Tuple) (*Segment, *Block, error) {
	segmentID := ts.selectSegment(tuple)
	segment := ts.segments[segmentID]
	segment.mu.RLock()
	defer segment.mu.RUnlock()

	blockID, err := segment.selectBlock(tuple)
	if err != nil {
		return nil, nil, err
	}
	block := segment.Blocks[blockID]

	return segment, block, nil
}

func (ts *TupleStore) selectSegment(tuple *Tuple) string {
	if !tuple.Timestamp.IsZero() && len(ts.timeSegments) > 0 {
		for _, seg := range ts.timeSegments {
			if seg.created.Year() == tuple.Timestamp.Year() {
				return seg.ID
			}
		}
	}

	if len(ts.idRangeSegments) > 0 {
		for _, seg := range ts.idRangeSegments {
			if len(tuple.ID) > 0 && seg.ID[0] == tuple.ID[0] {
				return seg.ID
			}
		}
	}

	hash := xxhash.Sum64String(tuple.ID)
	segmentIDs := make([]string, 0, len(ts.segments))
	for id := range ts.segments {
		segmentIDs = append(segmentIDs, id)
	}
	return segmentIDs[hash%uint64(len(segmentIDs))]
}

func (s *Segment) selectBlock(tuple *Tuple) (string, error) {
	s.mu.RLock()
	defer s.mu.RUnlock()

	if len(s.Blocks) == 0 {
		return "", fmt.Errorf("no blocks available in segment %s", s.ID)
	}

	hash := xxhash.Sum64String(tuple.ID)
	blockIDs := make([]string, 0, len(s.Blocks))
	for id := range s.Blocks {
		blockIDs = append(blockIDs, id)
	}
	return blockIDs[hash%uint64(len(blockIDs))], nil
}

func (ts *TupleStore) shouldMerge(block *Block) bool {
	block.mu.RLock()
	defer block.mu.RUnlock()

	if block.DeltaStore.ChangeCount >= ts.mergeThreshold {
		return true
	}

	if block.DeltaStore.Size() >= ts.maxDeltaSize {
		return true
	}

	if time.Since(block.LastMergeTime) > ts.maxMergeInterval {
		return true
	}

	return false
}

func (ts *TupleStore) scheduleBlockMerge(segmentID, blockID string) {
	start := time.Now()
	defer func() {
		ts.metrics.MergeDuration.Observe(time.Since(start).Seconds())
		ts.metrics.MergeCount.Inc()
	}()

	ts.mu.RLock()
	segment, exists := ts.segments[segmentID]
	ts.mu.RUnlock()

	if !exists {
		return
	}

	segment.mu.Lock()
	block, exists := segment.Blocks[blockID]
	segment.mu.Unlock()

	if !exists {
		return
	}

	block.mu.Lock()
	defer block.mu.Unlock()

	block.mergeDeltaStore()
	block.LastMergeTime = time.Now()
}

func (ts *TupleStore) updateIndices(tuple *Tuple, segmentID, blockID string) {
	ts.idIndex.Insert(tuple.ID, segmentID, blockID)
}
